package com.example.service.message.service;


import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.common.ResponseVO;
import com.example.common.constant.Constants;
import com.example.common.enums.ConversationTypeEnum;
import com.example.common.enums.DelFlagEnum;
import com.example.common.enums.MessageErrorCode;
import com.example.common.enums.command.Command;
import com.example.common.enums.command.GroupEventCommand;
import com.example.common.enums.command.MessageCommand;
import com.example.common.model.ClientInfo;
import com.example.common.model.SyncReq;
import com.example.common.model.SyncResp;
import com.example.common.model.message.MessageReadedContent;
import com.example.common.model.message.MessageReciveAckContent;
import com.example.common.model.message.OfflineMessageContent;
import com.example.common.model.message.RecallMessageContent;
import com.example.pack.message.MessageReadedPack;
import com.example.pack.message.RecallMessageNotifyPack;
import com.example.service.conversation.service.ConversationService;
import com.example.service.group.service.ImGroupMemberService;
import com.example.service.message.dao.ImMessageBodyEntity;
import com.example.service.message.dao.mapper.ImMessageBodyMapper;
import com.example.service.seq.RedisSeq;
import com.example.service.utils.ConversationIdGenerate;
import com.example.service.utils.GroupMessageProducer;
import com.example.service.utils.MessageProducer;
import com.example.service.utils.SnowflakeIdWorker;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;

@Service
public class MessageSyncService {

    @Autowired
    MessageProducer messageProducer;

    @Autowired
    ConversationService conversationService;

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    ImMessageBodyMapper imMessageBodyMapper;

    @Autowired
    RedisSeq redisSeq;

    @Autowired
    ImGroupMemberService imGroupMemberService;

    @Autowired
    GroupMessageProducer groupMessageProducer;

    public void receiveMark(MessageReciveAckContent messageReciveAckContent){
        messageProducer.sendToUser(messageReciveAckContent.getToId(), MessageCommand.MSG_RECIVE_ACK,messageReciveAckContent,messageReciveAckContent.getAppId());
    }


    //消息已读。更新会话的seq,通知在线的同步端发送指定command ,通知对方 我已读
    public void readMark(MessageReadedContent messageReadedContent) {
        //进行会话的更新或创建
        conversationService.messageMarkRead(messageReadedContent);

        MessageReadedPack messageReadedPack=new MessageReadedPack();     //消息已读的包装
        BeanUtils.copyProperties(messageReadedContent,messageReadedPack);
        //发送给自己的其他端
        syncToSender(messageReadedPack,messageReadedContent,MessageCommand.MSG_READED_NOTIFY);
        //发送给对方的所有端
        messageProducer.sendToUser(messageReadedContent.getToId(),MessageCommand.MSG_READED_RECEIPT,messageReadedPack,messageReadedContent.getAppId());
    }


    //发送给自己的其他段
    private void syncToSender(MessageReadedPack pack,MessageReadedContent content,Command command){
//        MessageReadedPack messageReadedPack=new MessageReadedPack();     //消息已读的包装

        //发送给自己的其他端  这里的getFromId  就是自己的id
        messageProducer.sendToUserExceptClient(pack.getFromId(),
                command,pack,content);

    }



    //群消息已读 标记
    public void groupReadMark(MessageReadedContent messageReaded) {
        conversationService.messageMarkRead(messageReaded);
        MessageReadedPack messageReadedPack = new MessageReadedPack();
        BeanUtils.copyProperties(messageReaded,messageReadedPack);
        //发送给自己的其他端
        syncToSender(messageReadedPack,messageReaded, GroupEventCommand.MSG_GROUP_READED_NOTIFY
        );
        //发送给群里的成员（除了自己）
        if(!messageReaded.getFromId().equals(messageReaded.getToId())){
            messageProducer.sendToUser(messageReadedPack.getToId(),GroupEventCommand.MSG_GROUP_READED_RECEIPT
                    ,messageReaded,messageReaded.getAppId());
        }
    }

    public ResponseVO syncOfflineMessage(SyncReq req) {
        SyncResp<OfflineMessageContent> resp = new SyncResp<>();
        String key = req.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + req.getOperater();
        //获取最大的seq
        Long maxSeq = 0L;
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();          //进行创建一个操作  这个属于一个api 可以调用一些方法
        Set set = zSetOperations.reverseRangeWithScores(key, 0, 0);    //进行获取最大的
        if(!CollectionUtils.isEmpty(set)){
            List list = new ArrayList<>(set);
            DefaultTypedTuple o= (DefaultTypedTuple) list.get(0);   //这个 DefaultTypedTuple是redis里面自带的
            maxSeq=o.getScore().longValue();                        //获得值最高的seq 方便后面做比较
        }

        List<OfflineMessageContent> respList=new ArrayList<>();
        resp.setMaxSequence(maxSeq);

        //进行获取离线消息 并进行分页
        Set<ZSetOperations.TypedTuple> queryset = zSetOperations.rangeByScoreWithScores(key, req.getLastSequence(), maxSeq, 0, req.getMaxLimit());
        for (ZSetOperations.TypedTuple<String> typedTuple : queryset) {
            String value = typedTuple.getValue();
            OfflineMessageContent offlineMessageContent = JSONObject.parseObject(value, OfflineMessageContent.class);
            respList.add(offlineMessageContent);
        }
        resp.setDataList(respList);
        //进行设置是否拉取完毕
        if (!CollectionUtils.isEmpty(respList)){
            OfflineMessageContent offlineMessageContent = respList.get(respList.size() - 1);
            resp.setCompleted(offlineMessageContent.getMessageSequence()>=maxSeq);
        }else {
            resp.setCompleted(true);
        }

        return ResponseVO.successResponse(resp);
    }

    //消息撤回的处理
    //修改离线消息的状态
    //ack给发送方
    //发送给同步端
    //分发给消息的接收方
    public void recallMessage(RecallMessageContent content) {
        //进行获取消息发起的时间
        Long messageTime = content.getMessageTime();
        //获取当前的时间
        Long now = System.currentTimeMillis();

        RecallMessageNotifyPack pack = new RecallMessageNotifyPack();
        BeanUtils.copyProperties(content, pack);

        if (120000L < now - messageTime) {     //超过两分钟的消息不能撤回
            recallAck(pack, ResponseVO.errorResponse(MessageErrorCode.MESSAGE_RECALL_TIME_OUT), content);
            return;
        }

        //进行查询消息
        QueryWrapper<ImMessageBodyEntity> query = new QueryWrapper<>();
        query.eq("app_id", content.getAppId());
        query.eq("message_key", content.getMessageKey());
        ImMessageBodyEntity body = imMessageBodyMapper.selectOne(query);


        if (body == null) {
            // ack失败 不存在的消息不能撤回
            recallAck(pack, ResponseVO.errorResponse(MessageErrorCode.MESSAGEBODY_IS_NOT_EXIST), content);
            return;
        }

        if (body.getDelFlag() == DelFlagEnum.DELETE.getCode()) {
            //已经撤回的消息 不能在继续撤回
            recallAck(pack, ResponseVO.errorResponse(MessageErrorCode.MESSAGE_IS_RECALLED), content);
            return;
        }

        //进行更新消息
        body.setDelFlag(DelFlagEnum.DELETE.getCode());
        imMessageBodyMapper.update(body, query);

        //下面是对离线消息的处理

        if (content.getConversationType() == ConversationTypeEnum.P2P.getCode()) {
            // 找到fromId的队列
            String fromKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getFromId();
            // 找到toId的队列
            String toKey = content.getAppId() + ":" + Constants.RedisConstants.OfflineMessage + ":" + content.getToId();

            //对离线消息的处理
            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
            BeanUtils.copyProperties(content, offlineMessageContent);
            offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
            offlineMessageContent.setMessageKey(content.getMessageKey());
            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
            offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType()
                    , content.getFromId(), content.getToId()));
            offlineMessageContent.setMessageBody(body.getMessageBody());

            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(), content.getToId()));
            offlineMessageContent.setMessageSequence(seq);      //给消息添加标识

            long messageKey = SnowflakeIdWorker.nextId();

            //放到离线消息里面
            redisTemplate.opsForZSet().add(fromKey, JSONObject.toJSONString(offlineMessageContent), messageKey);
            redisTemplate.opsForZSet().add(toKey, JSONObject.toJSONString(offlineMessageContent), messageKey);

            //ack
            recallAck(pack, ResponseVO.successResponse(), content);
            //分发给同步端
            messageProducer.sendToUserExceptClient(content.getFromId(),
                    MessageCommand.MSG_RECALL_NOTIFY, pack, content);
            //分发给对方
            messageProducer.sendToUser(content.getToId(), MessageCommand.MSG_RECALL_NOTIFY,
                    pack, content.getAppId());
        } else {
            List<String> groupMemberId = imGroupMemberService.getGroupMemberId(content.getToId(), content.getAppId());
            long seq = redisSeq.doGetSeq(content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(content.getFromId(), content.getToId()));
            //ack
            recallAck(pack, ResponseVO.successResponse(), content);
            //发送给同步端
            messageProducer.sendToUserExceptClient(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack
                    , content);
            for (String memberId : groupMemberId) {
                String toKey = content.getAppId() + ":" + Constants.SeqConstants.Message + ":" + memberId;
                OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
                offlineMessageContent.setDelFlag(DelFlagEnum.DELETE.getCode());
                BeanUtils.copyProperties(content, offlineMessageContent);
                offlineMessageContent.setConversationType(ConversationTypeEnum.GROUP.getCode());
                offlineMessageContent.setConversationId(conversationService.convertConversationId(offlineMessageContent.getConversationType()
                        , content.getFromId(), content.getToId()));
                offlineMessageContent.setMessageBody(body.getMessageBody());
                offlineMessageContent.setMessageSequence(seq);
                redisTemplate.opsForZSet().add(toKey, JSONObject.toJSONString(offlineMessageContent), seq);

                groupMessageProducer.producer(content.getFromId(), MessageCommand.MSG_RECALL_NOTIFY, pack, content);
            }
        }

    }



    private void recallAck(RecallMessageNotifyPack recallPack, ResponseVO<Object> success, ClientInfo clientInfo) {
        ResponseVO<Object> wrappedResp = success;
        messageProducer.sendToUser(recallPack.getFromId(),
                MessageCommand.MSG_RECALL_ACK, wrappedResp, clientInfo);
    }
}
